package producer

import (
	"github.com/Shopify/sarama"
	"log"
	"strings"
)

type KafkaProducer interface {
	Init()
	Close() error
	SendMsg(msg []byte)
}

func New(conf Config) KafkaProducer {
	return &producer{
		Config:      conf,
		closeSignal: make(chan bool),
		msgChan:     make(chan []byte),
	}
}

type producer struct {
	Producer sarama.AsyncProducer
	Config
	closeSignal chan bool
	msgChan     chan []byte
}

type Config struct {
	Hosts      string
	Topic      string
	Key        string
	Partition  int
	SaslEnable bool
	UserName   string
	Password   string
}

func (this *producer) Init() {
	log.Println("启动kafka生产者")
	if err := this.newProducer(); err != nil {
		panic(err.Error())
	}
	go func() {
		this.waitSendMsg()
	}()
	log.Println("kafka生产者已启动")
}

func (this *producer) SendMsg(msg []byte) {
	this.msgChan <- msg
}

func (this *producer) waitSendMsg() {
	log.Println("kafka生产者消息等待中")
	defer this.Producer.Close()
	defer log.Println("kafka生产者消息等待退出")
	for {
		select {
		case msg := <-this.msgChan:
			this.Producer.Input() <- &sarama.ProducerMessage{Topic: this.Topic, Key: sarama.StringEncoder(this.Key), Value: sarama.ByteEncoder(msg)}
		case <-this.closeSignal:
			return
		}
	}
}

func (this *producer) Close() error {
	defer log.Println("kafka生产者退出")
	this.closeSignal <- true
	return nil
}

func (this *producer) newProducer() (err error) {
	//创建配置
	config := sarama.NewConfig()
	if this.SaslEnable {
		config.Net.SASL.Enable = true
		config.Net.SASL.User = this.UserName
		config.Net.SASL.Password = this.Password
	}
	//创建客户端
	client, err := sarama.NewClient(strings.Split(this.Hosts, ","), config)
	if err != nil {
		return err
	}
	//创建生产者
	this.Producer, err = sarama.NewAsyncProducerFromClient(client)
	return err
}
